package com.xukang.batch.config;

import com.xukang.batch.listener.MyJobListener;
import com.xukang.batch.listener.MyStepListener;
import com.xukang.batch.po.Users;
import com.xukang.batch.processor.*;
import com.xukang.batch.service.UserService;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepScope;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.repository.support.JobRepositoryFactoryBean;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.transaction.PlatformTransactionManager;

import javax.annotation.Resource;
import javax.sql.DataSource;

@Configuration
@EnableBatchProcessing
public class BatchConfig {

    @Autowired
    private JobBuilderFactory jobBuilderFactory;
    @Autowired
    private StepBuilderFactory stepBuilderFactory;
    @Resource
    private UserService userService;

    /**
     * @description: 定义Job
     */
    @Bean
    public Job importUserJob() {
        return jobBuilderFactory.get("importUserJob")
                .start(processingStep())
                .listener(new MyJobListener())
//                .add(step2(), step3())
                .build();
    }

    /**
     * @description: 定义Step
     */
    @Bean
    public Step processingStep() {

        return stepBuilderFactory.get("tabelStep")
                .listener(new MyStepListener()) // 自定义监听器
                // 每处理10条或处理完 write 一次
                .<Users, Users>chunk(10)
                .reader( itemReader1())
                .processor(itemProcessor())
                .writer(itemWriter())

//                .faultTolerant()//配置步骤的容错特性
//                .skipPolicy(new AlwaysSkipItemSkipPolicy()) // 总是跳过
//                .retryLimit(3) // 重试次数
//                .retry(ConnectException.class) // 网络问题重试
//                .retry(DeadlockLoserDataAccessException.class) // 数据库死锁重试
//                .skipLimit(3) // 最多跳过3条错误
////              .skip(IllegalArgumentException.class) // 该异常跳过
//                .noRollback(ValidationException.class) // 验证异常不回滚

                .build();
// 分区处理器实现
//        return stepBuilderFactory.get("masterStep")
//                .partitioner("slaveStep", partitioner())
//                .gridSize(10) // 分区数量=CPU核心数*2
//                .taskExecutor(new ThreadPoolTaskExecutor())
//                .build();

    }

    @Bean
    @StepScope
    public ItemReader<Users> itemReader() {
        //读取方案: 使用@StepScope和@JobScope  通过 Spring 的作用域确保每次启动获取新实例或参数
        return new MyItemReader(userService);
    }

    @Bean
    public ItemReader<Users> itemReader1() {
        // 读取方案1: 通过实现ItemStreamReader跟踪读取位置，Spring Batch 会自动恢复状态, 读取位置应当保存到 数据库中
        return new MyItemReader1();
    }


    @Bean
    public MyItemReader2 itemReader2() {
        // 读取方案2: 从作业参数或数据库获取执行参数，一次执行，反复拉取，直到全部执行完
        return new MyItemReader2();
    }

    @Bean
    public ItemProcessor<Users, Users> itemProcessor() {
        return new MyItemProcessor();
    }

    @Bean
    public ItemWriter<Users> itemWriter() {
        return new MyItemWriter();
    }

    /**
     * 启用作业元数据存储
     */

    @Bean("myJobRepository")
    public JobRepository myJobRepository(DataSource dataSource, PlatformTransactionManager transactionManager) throws Exception {
        // 便于 ExecutionContext 持久化,保留上次执行位置
        JobRepositoryFactoryBean factory = new JobRepositoryFactoryBean();
        factory.setDataSource(dataSource);
        factory.setTransactionManager(transactionManager);
        return factory.getObject();
    }
}